package client

import (
	"fmt"
	"io"
	"net"
	"runtime"
	"sync"
	"sync/atomic"
	"time"

	"gitee.com/sdxstar/air"
	"gitee.com/sdxstar/air/logger"
	"gitee.com/sdxstar/air/options"
	"gitee.com/sdxstar/air/protocol"

	"github.com/pkg/errors"
)

type EventHandler func(c Context) error

type Client struct {
	opts *options.ClientOptions
	conn net.Conn

	address     string
	idGenerator *air.IdGenerator

	eventHandlers     map[air.Event]EventHandler
	eventHandlersLock sync.RWMutex

	msgHandlers     map[string]EventHandler
	msgHandlersLock sync.RWMutex

	closeFlag int32
	closeChan chan struct{}

	heartbeatResetChan chan struct{}
	receiveChan        chan protocol.Packet
	sendChan           chan protocol.Packet

	waitReply sync.Map

	disconnectChan chan struct{}
	reconnectCount uint32

	waitGroup sync.WaitGroup

	log logger.Logger
}

func NewClient(opts *options.ClientOptions) (c *Client, err error) {
	c = &Client{
		opts:          opts,
		waitReply:     sync.Map{},
		eventHandlers: make(map[air.Event]EventHandler),
		msgHandlers:   make(map[string]EventHandler),
	}
	c.buildOpts()
	c.idGenerator, err = air.NewSnowFlake(c.opts.WorkId)
	return
}

func (c *Client) addWaitReplyChan(id uint64, waitReplyChan chan []byte) {
	c.waitReply.Store(id, waitReplyChan)
}

func (c *Client) removeWaitReplyChan(id uint64) {
	c.waitReply.Delete(id)
}

func (c *Client) buildOpts() {
	defaultOpt := options.DefaultClientOptions()
	if c.opts == nil {
		c.opts = defaultOpt
	}
	if c.opts.WorkId == 0 {
		c.opts.WorkId = defaultOpt.WorkId
	}
	if c.opts.AddressFetcher == nil {
		c.opts.AddressFetcher = defaultOpt.AddressFetcher
	}
	if c.opts.HeartbeatInterval == 0 {
		c.opts.HeartbeatInterval = defaultOpt.HeartbeatInterval
	}
	if c.opts.MessageTimeout == 0 {
		c.opts.MessageTimeout = defaultOpt.MessageTimeout
	}
	if c.opts.PacketSendChanLimit == 0 {
		c.opts.PacketSendChanLimit = defaultOpt.PacketSendChanLimit
	}
	if c.opts.PacketReceiveChanLimit == 0 {
		c.opts.PacketReceiveChanLimit = defaultOpt.PacketReceiveChanLimit
	}
	if c.opts.Codec == nil {
		c.opts.Codec = defaultOpt.Codec
	}
	if c.opts.Protocol == nil {
		c.opts.Protocol = defaultOpt.Protocol
	}
	if c.opts.Logger == nil {
		c.opts.Logger = logger.NewConsoleLog()
	}
	c.log = c.opts.Logger
}

func (c *Client) MustStart() {
	go func() {
		for {
			if err := c.Start(); err != nil {
				if c.IsClosed() {
					return
				}
				c.log.Error("connection establish failed: " + err.Error())
				c.log.Info(fmt.Sprintf("client will retry to connect in %s", c.opts.ReconnectInterval))
				time.Sleep(c.opts.ReconnectInterval)
				continue
			}
			return
		}
	}()
}

func (c *Client) Start() error {
	if err := c.FetchAddress(); err != nil {
		return err
	}
	conn, err := net.Dial("tcp", c.address)
	if err != nil {
		return errors.Wrapf(err, "air client dial to %s failed", c.address)
	}
	c.log.Info(fmt.Sprintf("client connected to %s", c.address))
	c.reconnectCount = 0
	c.conn = conn
	c.heartbeatResetChan = make(chan struct{})
	c.closeChan = make(chan struct{})
	c.receiveChan = make(chan protocol.Packet, c.opts.PacketReceiveChanLimit)
	c.sendChan = make(chan protocol.Packet, c.opts.PacketSendChanLimit)
	c.disconnectChan = make(chan struct{})

	go c.handleLoop()
	c.process()

	if c.opts.Reconnect {
		c.reconnect()
	} else {
		c.Close()
	}
	return nil
}

func (c *Client) FetchAddress() (err error) {
	c.address, err = c.opts.AddressFetcher()
	if err != nil {
		c.log.Error("fetch address failed: ", err)
	} else if c.address == "" {
		err = errors.New("empty address")
		c.log.Error("fetch address failed: ", err)
	}
	return
}

func (c *Client) reconnect() {
	if c.IsClosed() {
		return
	}
	if c.opts.ReconnectLimit > 0 && c.reconnectCount > c.opts.ReconnectLimit {
		c.log.Error(fmt.Sprintf("client retry reconnect %d times, giveup", c.reconnectCount))
		return
	}
	c.log.Info(fmt.Sprintf("client will reconnect in %s", c.opts.ReconnectInterval))
	if h, ok := c.eventHandlers[air.OnReconnect]; ok {
		go func() {
			defer func() {
				if r := recover(); r != nil {
					err, ok := r.(error)
					if !ok {
						err = fmt.Errorf("%v", r)
					}
					stack := make([]byte, 4<<10)
					length := runtime.Stack(stack, true)
					msg := fmt.Sprintf("[PANIC RECOVER] %v %s\n", err, stack[:length])
					c.log.Error(msg)
				}
			}()
			err := h(newContext(c, nil))
			if err != nil {
				c.log.Error(err)
			}
		}()
	}
	t := time.After(c.opts.ReconnectInterval)
	select {
	case <-c.closeChan:
		c.Close()
	case <-t:
		if c.IsClosed() {
			return
		}
		if err := c.FetchAddress(); err != nil {
			c.reconnectCount++
			c.reconnect()
			return
		}
		if c.IsClosed() {
			return
		}

		conn, err := net.Dial("tcp", c.address)
		if err != nil {
			c.reconnectCount++
			c.log.Error(err)
			c.reconnect()
			return
		}
		c.log.Tracef("client connected to %s", c.address)
		c.reconnectCount = 0
		c.disconnectChan = make(chan struct{})
		c.conn = conn
		c.process()
		c.reconnect()
		return
	}
}

func (c *Client) process() {
	c.waitGroup.Add(3)
	go c.heartbeatLoop()
	go c.readLoop()
	go c.writeLoop()

	if h, ok := c.eventHandlers[air.OnConnect]; ok {
		go func() {
			defer func() {
				if r := recover(); r != nil {
					err, ok := r.(error)
					if !ok {
						err = fmt.Errorf("%v", r)
					}
					stack := make([]byte, 4<<10)
					length := runtime.Stack(stack, true)
					msg := fmt.Sprintf("[PANIC RECOVER] %v %s\n", err, stack[:length])
					c.log.Error(msg)
				}
			}()
			err := h(newContext(c, nil))
			if err != nil {
				c.log.Error(err)
				c.Close()
			}
		}()
	}
	c.waitGroup.Wait()
	if h, ok := c.eventHandlers[air.OnDisconnect]; ok {
		go func() {
			defer func() {
				if r := recover(); r != nil {
					err, ok := r.(error)
					if !ok {
						err = fmt.Errorf("%v", r)
					}
					stack := make([]byte, 4<<10)
					length := runtime.Stack(stack, true)
					msg := fmt.Sprintf("[PANIC RECOVER] %v %s\n", err, stack[:length])
					c.log.Error(msg)
				}
			}()
			err := h(newContext(c, nil))
			if err != nil {
				c.log.Error(err)
			}
		}()
	}
}

func (c *Client) Close() {
	if c.IsClosed() {
		return
	}
	atomic.StoreInt32(&c.closeFlag, 1)
	if c.closeChan != nil {
		close(c.closeChan)
	}
	if c.receiveChan != nil {
		close(c.receiveChan)
	}
	if c.sendChan != nil {
		close(c.sendChan)
	}
	if c.heartbeatResetChan != nil {
		close(c.heartbeatResetChan)
	}
	if c.conn != nil {
		_ = c.conn.Close()
	}
	if h, ok := c.eventHandlers[air.OnClose]; ok {
		go func() {
			defer func() {
				if r := recover(); r != nil {
					err, ok := r.(error)
					if !ok {
						err = fmt.Errorf("%v", r)
					}
					stack := make([]byte, 4<<10)
					length := runtime.Stack(stack, true)
					msg := fmt.Sprintf("[PANIC RECOVER] %v %s\n", err, stack[:length])
					c.log.Error(msg)
				}
			}()
			err := h(newContext(c, nil))
			if err != nil {
				c.log.Error(err)
			}
		}()
	}
	c.log.Info("client closed")
}

func (c *Client) IsClosed() bool {
	return atomic.LoadInt32(&c.closeFlag) == 1
}

func (c *Client) RemoteAddr() string {
	if c.conn == nil {
		return c.address
	}
	return c.conn.RemoteAddr().String()
}

func (c *Client) On(e string, fn EventHandler) {
	switch e {
	case air.OnConnect, air.OnHeartbeat, air.OnJoin, air.OnLeave, air.OnDisconnect, air.OnReconnect, air.OnClose:
		c.eventHandlersLock.Lock()
		defer c.eventHandlersLock.Unlock()
		c.eventHandlers[e] = fn
	default:
		c.msgHandlersLock.Lock()
		defer c.msgHandlersLock.Unlock()
		c.msgHandlers[e] = fn
	}
}

func (c *Client) heartbeatLoop() {
	defer func() {
		if r := recover(); r != nil {
			err, ok := r.(error)
			if !ok {
				err = fmt.Errorf("%v", r)
			}
			stack := make([]byte, 4<<10)
			length := runtime.Stack(stack, true)
			msg := fmt.Sprintf("[PANIC RECOVER] %v %s\n", err, stack[:length])
			c.log.Error(msg)
		}
		c.log.Trace("exit heartbeatLoop")
		c.waitGroup.Done()
	}()
	t := time.NewTicker(c.opts.HeartbeatInterval)
	for {
		select {
		case <-t.C:
			c.sendChan <- c.opts.Protocol.NewPacket(protocol.MessageKeyHeartbeat, 0, "", nil)
		case <-c.heartbeatResetChan:
			t.Reset(c.opts.HeartbeatInterval)
			continue
		case <-c.disconnectChan:
			return
		case <-c.closeChan:
			return
		}
	}
}

func (c *Client) readLoop() {
	defer func() {
		if r := recover(); r != nil {
			err, ok := r.(error)
			if !ok {
				err = fmt.Errorf("%v", r)
			}
			stack := make([]byte, 4<<10)
			length := runtime.Stack(stack, true)
			msg := fmt.Sprintf("[PANIC RECOVER] %v %s\n", err, stack[:length])
			c.log.Error(msg)
		}
		c.log.Tracef("exit readLoop")
		c.waitGroup.Done()
	}()

	for {
		select {
		case <-c.closeChan:
			c.log.Tracef("close disconnectChan")
			close(c.disconnectChan)
			return
		default:
		}

		p, err := c.opts.Protocol.ReadPacket(c.conn)
		if err != nil {
			if errors.Cause(err) == io.EOF {
				c.log.Tracef("server closed connection")
			} else {
				c.log.Error(err)
			}
			c.log.Tracef("close disconnectChan")
			close(c.disconnectChan)
			return
		}
		c.receiveChan <- p
	}
}

func (c *Client) writeLoop() {
	defer func() {
		if r := recover(); r != nil {
			err, ok := r.(error)
			if !ok {
				err = fmt.Errorf("%v", r)
			}
			stack := make([]byte, 4<<10)
			length := runtime.Stack(stack, true)
			msg := fmt.Sprintf("[PANIC RECOVER] %v %s\n", err, stack[:length])
			c.log.Error(msg)
		}
		c.log.Tracef("exit writeLoop")
		c.waitGroup.Done()
	}()

	for {
		select {
		case <-c.closeChan:
			return
		case <-c.disconnectChan:
			return
		case p := <-c.sendChan:
			if c.IsClosed() {
				return
			}
			c.heartbeatResetChan <- struct{}{}
			if _, err := c.opts.Protocol.WritePacket(p, c.conn); err != nil {
				c.log.Error(err)
				c.conn.Close()
				return
			}
		}
	}
}

func (c *Client) handleLoop() {
	defer func() {
		if r := recover(); r != nil {
			err, ok := r.(error)
			if !ok {
				err = fmt.Errorf("%v", r)
			}
			stack := make([]byte, 4<<10)
			length := runtime.Stack(stack, true)
			msg := fmt.Sprintf("[PANIC RECOVER] %v %s\n", err, stack[:length])
			c.log.Error(msg)
		}
		c.log.Tracef("exit handleLoop")
	}()

	for {
		select {
		case <-c.closeChan:
			return
		case p := <-c.receiveChan:
			if c.IsClosed() {
				return
			}

			switch p.Key() {
			case protocol.MessageKeyHeartbeat:
			case protocol.MessageKeyEmit, protocol.MessageKeyAckRequest:
				go func() {
					defer func() {
						if r := recover(); r != nil {
							err, ok := r.(error)
							if !ok {
								err = fmt.Errorf("%v", r)
							}
							stack := make([]byte, 4<<10)
							length := runtime.Stack(stack, true)
							msg := fmt.Sprintf("[PANIC RECOVER] %v %s\n", err, stack[:length])
							c.log.Error(msg)
						}
					}()
					if h, ok := c.msgHandlers[p.Event()]; ok {
						err := h(newContext(c, p))
						if err != nil {
							c.log.Error(err)
						}
					} else {
						c.log.Error("not find handler for event ", p.Event())
					}
				}()
			case protocol.MessageKeyAckResponse:
				go func() {
					if replyChan, ok := c.waitReply.Load(p.Id()); ok {
						replyChan.(chan []byte) <- p.Body()
					}
				}()
			}

		}
	}
}

func (c *Client) reply(p protocol.Packet) (err error) {
	if c.IsClosed() {
		return air.ErrConnClosed
	}

	defer func() {
		if r := recover(); r != nil {
			err, ok := r.(error)
			if !ok {
				err = fmt.Errorf("%v", r)
			}
			stack := make([]byte, 4<<10)
			length := runtime.Stack(stack, true)
			msg := fmt.Sprintf("[PANIC RECOVER] %v %s\n", err, stack[:length])
			c.log.Error(msg)
		}
	}()

	select {
	case c.sendChan <- p:
		return nil
	case <-c.closeChan:
		return air.ErrConnClosed
	case <-time.After(c.opts.MessageTimeout):
		return air.ErrMessageTimeout
	}
}

func (c *Client) Emit(event string, v interface{}) (err error) {
	if c.IsClosed() {
		return air.ErrConnClosed
	}

	defer func() {
		if r := recover(); r != nil {
			err, ok := r.(error)
			if !ok {
				err = fmt.Errorf("%v", r)
			}
			stack := make([]byte, 4<<10)
			length := runtime.Stack(stack, true)
			msg := fmt.Sprintf("[PANIC RECOVER] %v %s\n", err, stack[:length])
			c.log.Error(msg)
		}
	}()

	var p protocol.Packet

	if v == nil {
		p = c.opts.Protocol.NewPacket(protocol.MessageKeyEmit, c.idGenerator.Generate(), event, nil)
	} else {
		msg, err := c.opts.Codec.Encode(v)
		if err != nil {
			return err
		}
		p = c.opts.Protocol.NewPacket(protocol.MessageKeyEmit, c.idGenerator.Generate(), event, msg)
	}

	select {
	case c.sendChan <- p:
		return nil
	case <-c.closeChan:
		return air.ErrConnClosed
	case <-time.After(5 * time.Second):
		return air.ErrMessageTimeout
	}
}

func (c *Client) Ack(event string, v interface{}, r interface{}) (err error) {
	if c.IsClosed() {
		return air.ErrConnClosed
	}

	defer func() {
		if r := recover(); r != nil {
			err, ok := r.(error)
			if !ok {
				err = fmt.Errorf("%v", r)
			}
			stack := make([]byte, 4<<10)
			length := runtime.Stack(stack, true)
			msg := fmt.Sprintf("[PANIC RECOVER] %v %s\n", err, stack[:length])
			c.log.Error(msg)
		}
	}()

	var p protocol.Packet

	if v == nil {
		p = c.opts.Protocol.NewPacket(protocol.MessageKeyAckRequest, c.idGenerator.Generate(), event, nil)
	} else {
		msg, err := c.opts.Codec.Encode(v)
		if err != nil {
			return err
		}
		p = c.opts.Protocol.NewPacket(protocol.MessageKeyAckRequest, c.idGenerator.Generate(), event, msg)
	}

	replyChan := make(chan []byte)
	c.addWaitReplyChan(p.Id(), replyChan)

	c.sendChan <- p

	select {
	case reply := <-replyChan:
		c.removeWaitReplyChan(p.Id())
		close(replyChan)
		return c.opts.Codec.Decode(reply, r)
	case <-c.closeChan:
		return air.ErrConnClosed
	case <-time.After(c.opts.MessageTimeout):
		c.removeWaitReplyChan(p.Id())
		close(replyChan)
		return air.ErrMessageTimeout
	}
}
